"""
Case Type   : 备机支持逻辑复制
Case Name   : 一主两备环境下，两个备机同时对同一个逻辑复制槽解码
Create At   : 2024/07/23
Owner       : lvlintao666
Description :
    step1:修改参数wal_level为logical;enable_slot_log为on;
    step2:创建逻辑复制用户;
    step3:配置逻辑复制用户白名单;
    step4:创建逻辑复制槽;
    step5:查询复制槽;
    step6:备机1执行逻辑复制槽流式解码;
    step7:备机2解码同一个逻辑复制槽;
    step8:创建表(无主键)并进行DML操作;
    step9:备机1停止解码，step6返回结果;
    step10:备机2停止解码，step7返回结果;
    step11:两个备机查看解码文件;
    step12:清理环境;expect:清理环境完成
Expect      :
    step1:修改参数wal_level为logical;enable_slot_log为on;expect:修改成功
    step2:创建逻辑复制用户;expect:创建成功
    step3:配置逻辑复制用户白名单;expect:设置成功
    step4:创建逻辑复制槽;expect:创建成功
    step5:查询复制槽;expect:显示slot_name复制槽信息
    step6:备机1执行逻辑复制槽流式解码;expect:开始解码，解码最终成功
    step7:备机2解码同一个逻辑复制槽;expect:解码报错
    step8:创建表(无主键)并进行DML操作;expect:创建成功
    step9:备机1停止解码，step6返回结果;expect:pg_recvlogical打印正常
    step10:备机2停止解码，step7返回结果;expect:pg_recvlogical打印正常
    step11:两个备机查看解码文件;expect:解码文件依然有解码信息，不会记录列的旧值
    step12:清理环境;expect:清理环境完成
History     :
    Modified by peilinqian at 2024-12-23:用例偶现gs_guc部分节点失败，增加日志打印及
        操作步骤预期与实际用例不符，优化用例逻辑
"""
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

Pri_SH = CommonSH('PrimaryDbUser')
node_num = Pri_SH.get_node_num()


@unittest.skipIf(3 > node_num, '非1主2备环境不执行')
class LogicalReplication(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.constant = Constant()
        self.com = Common()
        self.primary_node = Node('PrimaryDbUser')
        self.standby1_node = Node('Standby1DbUser')
        self.standby2_node = Node('Standby2DbUser')
        self.pg_hba = os.path.join(macro.DB_INSTANCE_PATH,
                                   macro.PG_HBA_FILE_NAME)
        self.parent_path = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.decode_file_01 = os.path.join(self.parent_path, 'decode01')
        self.decode_file_02 = os.path.join(self.parent_path, 'decode02')
        self.us_name = "u_logical_replication_0021"
        self.slot_name = "slot_logical_replication_0021"
        self.tb_name = "tb_logical_replication_0021"
        self.ha_port = str(int(self.primary_node.db_port) + 1)

    def test_standby(self):
        text = '----step1:修改参数wal_level为logical;enable_slot_log为on;' \
               'expect:修改成功----'
        self.log.info(text)
        guc_res1 = Pri_SH.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'wal_level=logical')
        self.log.info(guc_res1)
        self.assertTrue(guc_res1, '执行失败:' + text)
        guc_res2 = Pri_SH.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_slot_log=on')
        self.log.info(guc_res2)
        self.assertTrue(guc_res2, '执行失败:' + text)
        restart_msg = Pri_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = Pri_SH.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status,
                        '执行失败:' + text)

        text = '----step2:创建逻辑复制用户;expect:创建成功----'
        self.log.info(text)
        sql_cmd = Pri_SH.execut_db_sql(f"drop role if exists {self.us_name};"
                                       f"create role {self.us_name} "
                                       f"with login password "
                                       f"'{macro.COMMON_PASSWD}';"
                                       f"alter role {self.us_name} "
                                       f"with replication sysadmin;")
        self.log.info(sql_cmd)
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, sql_cmd,
                      '执行失败:' + text)

        text = '----step3:配置逻辑复制用户白名单;expect:设置成功----'
        self.log.info(text)
        guc_res3 = Pri_SH.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH,
            f'local   replication  {self.us_name}  trust')
        self.log.info(guc_res3)
        self.assertTrue(guc_res3, "执行失败" + text)
        guc_res4 = Pri_SH.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH,
            f'host  replication  {self.us_name}  127.0.0.1/32   trust')
        self.log.info(guc_res4)
        self.assertTrue(guc_res4, "执行失败" + text)
        guc_res5 = Pri_SH.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH,
            f'host   replication  {self.us_name}  ::1/128    trust')
        self.log.info(guc_res5)
        self.assertTrue(guc_res5, "执行失败" + text)

        text = '----step4:创建逻辑复制槽;expect:创建成功----'
        self.log.info(text)
        check_res = Pri_SH.execut_db_sql('select slot_name from '
                                         'pg_replication_slots;')
        self.log.info(check_res)
        if f'{self.slot_name}' in check_res.split('\n')[-2].strip():
            del_cmd = Pri_SH.execut_db_sql(f"select * from "
                                           f"pg_drop_replication_slot"
                                           f"('{self.slot_name}');")
            self.log.info(del_cmd)
        cre_cmd = Pri_SH.execut_db_sql(f"select * from "
                                       f"pg_create_logical_replication_slot"
                                       f"('{self.slot_name}', "
                                       f"'mppdb_decoding');")
        self.log.info(cre_cmd)
        self.assertNotIn('FATAL' and 'ERROR', cre_cmd,
                         '执行失败:' + text)

        text = f'----step5:查询复制槽;expect:显示"{self.slot_name}"复制槽信息----'
        self.log.info(text)
        query_cmd = Pri_SH.execut_db_sql("select slot_name,plugin from "
                                         "pg_get_replication_slots();"
                                         "select * from pg_replication_slots;")
        self.log.info(query_cmd)
        self.assertIn(f'{self.slot_name}', query_cmd, '执行失败:' + text)

        text = '----step6:备机1执行逻辑复制槽流式解码;expect:开始解码，解码最终成功----'
        self.log.info(text)
        decode_cmd = f"pg_recvlogical " \
                     f"-d {self.standby1_node.db_name} " \
                     f"-S {self.slot_name} " \
                     f"-p {self.ha_port} " \
                     f"--start " \
                     f"-f {self.decode_file_01} " \
                     f"-s 2 " \
                     f"-v " \
                     f"-P mppdb_decoding " \
                     f"-U {self.us_name}"
        execute_cmd = f'''source {macro.DB_ENV_PATH}
                          expect <<EOF
                          set timeout 300
                          spawn {decode_cmd}
                          expect "Password:"
                          send "{macro.COMMON_PASSWD}\\n"
                          expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        thread_1 = ComThread(self.com.get_sh_result, args=(self.standby1_node,
                                                           execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        time.sleep(3)

        text = '----step7:备机2解码同一个逻辑复制槽;expect:解码报错----'
        self.log.info(text)
        decode_cmd = f"pg_recvlogical " \
                     f"-d {self.standby2_node.db_name} " \
                     f"-S {self.slot_name} " \
                     f"-p {self.ha_port} " \
                     f"--start " \
                     f"-f {self.decode_file_02} " \
                     f"-s 2 " \
                     f"-v " \
                     f"-P mppdb_decoding " \
                     f"-U {self.us_name}"
        execute_cmd = f'''source {macro.DB_ENV_PATH}
                          expect <<EOF
                          set timeout 300
                          spawn {decode_cmd}
                          expect "Password:"
                          send "{macro.COMMON_PASSWD}\\n"
                          expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        thread_2 = ComThread(self.com.get_sh_result, args=(self.standby2_node,
                                                           execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()

        text = '----step8:创建表(无主键)并进行DML操作;expect:创建成功----'
        self.log.info(text)
        sql_cmd = Pri_SH.execut_db_sql(f'''drop table if exists {self.tb_name};
           create table {self.tb_name}(c_1 integer,
           c_2 bigint,
           c_3 smallint,
           c_4 tinyint,
           c_5 serial,
           c_6 smallserial,
           c_7 bigserial,
           c_8 float,
           c_9 double precision,
           c_10 DATE,
           c_11 time without time zone,
           c_12 timestamp without time zone,
           c_13 char(10),
           c_14 varchar(20),
           c_15 text,
           c_16 blob,
           c_17 bytea);
           insert into {self.tb_name} values(1,10,5,25,default,default,\
           default,1237.127,123456.1234,DATE '12-10-2010','21:21:21',\
           '2010-12-12', '测试','测试工程师','西安',empty_blob(),\
           E'\\xDEADBEEF');
           update {self.tb_name} set c_15 = '数据库';
           delete from {self.tb_name} where c_16 = empty_blob();''')
        self.log.info(sql_cmd)
        self.assertTrue('INSERT' in sql_cmd and 'UPDATE' in sql_cmd and
                        'DELETE' in sql_cmd, '执行失败:' + text)
        time.sleep(3)

        self.log.info('----等待主备同步----')
        Pri_SH.check_location_consistency('primary', node_num)

        text = '----step9:备机1停止解码，step6返回结果;expect:pg_recvlogical打印正常----'
        self.log.info(text)
        self.log.info('----停止解码,备机1解码pg_recvlogical打印结果------')
        stop_cmd = "ps -ef |  grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.log.info(stop_cmd)
        result = self.standby1_node.sh(stop_cmd).result()
        self.log.info(result)
        thread_1.join(60)
        thread_1_result = thread_1.get_result()
        self.assertIn('confirming write up', thread_1_result,
                      '执行失败:' + text)
        self.assertNotIn('FATAL' and 'ERROR', thread_1_result,
                         '执行失败:' + text)

        text = '----step10:备机2停止解码，step7返回结果;expect:pg_recvlogical打印正常----'
        self.log.info(text)
        self.log.info('----停止解码,备机1解码pg_recvlogical打印结果----')
        stop_cmd = "ps -ef |  grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.log.info(stop_cmd)
        result = self.standby2_node.sh(stop_cmd).result()
        self.log.info(result)
        thread_2.join(60)
        thread_2_result = thread_2.get_result()
        self.assertIn('confirming write up', thread_2_result,
                      '执行失败:' + text)
        self.assertNotIn('FATAL' and 'ERROR', thread_2_result,
                         '执行失败:' + text)

        text = '----step11:两个备机查看解码文件;expect:解码文件依然有解码信息，' \
               '不会记录列的旧值----'
        self.log.info(text)
        self.log.info('----备机1解码结果----')
        cat_cmd = f"cat {self.decode_file_01};"
        self.log.info(cat_cmd)
        result = self.standby1_node.sh(cat_cmd).result()
        self.log.info(result)
        self.assertIn('"old_keys_name":[]', result, '执行失败:' + text)

        self.log.info('----备机2解码结果----')
        cat_cmd = f"cat {self.decode_file_02};"
        self.log.info(cat_cmd)
        result = self.standby2_node.sh(cat_cmd).result()
        self.log.info(result)
        self.assertIn('"old_keys_name":[]', result, '执行失败:' + text)

    def tearDown(self):
        text = '----step12:清理环境;expect:清理环境完成----'
        self.log.info(text)
        self.log.info('删除用户和表')
        drop_res1 = Pri_SH.execut_db_sql(f"drop role if exists "
                                         f"{self.us_name};")
        self.log.info(drop_res1)
        drop_res2 = Pri_SH.execut_db_sql(f"drop table if exists "
                                         f"{self.tb_name};")
        self.log.info(drop_res2)
        self.log.info('删除复制槽')
        drop_slot_res = Pri_SH.execut_db_sql(f"select * from "
                                             f"pg_drop_replication_slot"
                                             f"('{self.slot_name}');")
        self.log.info(drop_slot_res)
        self.log.info('删除解码文件')
        rm_cmd1 = f"rm -rf {self.decode_file_01};"
        self.log.info(rm_cmd1)
        result1 = self.standby1_node.sh(rm_cmd1).result()
        self.log.info(result1)
        rm_cmd2 = f"rm -rf {self.decode_file_02};"
        self.log.info(rm_cmd2)
        result2 = self.standby2_node.sh(rm_cmd2).result()
        self.log.info(result2)

        self.log.info('-----gs_guc reload 失败增加日志打印-----')
        self.log.info('-----查看主机名-----')
        pri_node_name = self.com.get_sh_result(self.primary_node, 'uname -n')
        sta1_node_name = self.com.get_sh_result(self.standby1_node, 'uname -n')
        pssh_pri_cmd = f'pssh -s -H {pri_node_name} "hostname;' \
                       f'source {macro.DB_ENV_PATH};' \
                       f'gsql -r -c \\"select 1\\";"'
        pssh_sta1_cmd = f'pssh -s -H {sta1_node_name} "hostname;' \
                        f'source {macro.DB_ENV_PATH};' \
                        f'gsql -r -c \\"select 1\\";"'
        status = Pri_SH.get_db_cluster_status('detail')
        self.log.info(status)
        self.log.info('-----执行pssh命令-----')
        self.log.info(pssh_pri_cmd)
        self.com.get_sh_result(self.standby2_node, pssh_pri_cmd)
        self.log.info(pssh_sta1_cmd)
        self.com.get_sh_result(self.standby2_node, pssh_sta1_cmd)

        self.log.info('----恢复pg_hba.conf文件----')
        guc_res1 = Pri_SH.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH,
            f'local   replication  {self.us_name}  trust')
        self.log.info(guc_res1)
        self.assertTrue(guc_res1, "执行失败" + text)

        self.log.info('-----gs_guc reload 失败增加日志打印-----')
        status = Pri_SH.get_db_cluster_status('detail')
        self.log.info(status)
        self.log.info('-----执行pssh命令-----')
        self.log.info(pssh_pri_cmd)
        self.com.get_sh_result(self.standby2_node, pssh_pri_cmd)
        self.log.info(pssh_sta1_cmd)
        self.com.get_sh_result(self.standby2_node, pssh_sta1_cmd)

        guc_res2 = Pri_SH.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH,
            f'host  replication  {self.us_name}  127.0.0.1/32')
        self.log.info(guc_res2)
        self.assertTrue(guc_res2, "执行失败" + text)

        self.log.info('-----gs_guc reload 失败增加日志打印-----')
        status = Pri_SH.get_db_cluster_status('detail')
        self.log.info(status)
        self.log.info('-----执行pssh命令-----')
        self.log.info(pssh_pri_cmd)
        self.com.get_sh_result(self.standby2_node, pssh_pri_cmd)
        self.log.info(pssh_sta1_cmd)
        self.com.get_sh_result(self.standby2_node, pssh_sta1_cmd)

        guc_res3 = Pri_SH.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH,
            f'host   replication  {self.us_name}  ::1/128')
        self.log.info(guc_res3)

        self.log.info('----恢复参数默认值----')
        guc_res4 = Pri_SH.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'wal_level=hot_standby')
        self.log.info(guc_res4)
        guc_res5 = Pri_SH.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_slot_log=off')
        self.log.info(guc_res5)
        restart_msg = Pri_SH.restart_db_cluster()
        self.log.info(restart_msg)
        status = Pri_SH.get_db_cluster_status("detail")
        self.log.info(status)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, drop_res1,
                      '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_res2,
                      '执行失败:' + text)
        self.assertEqual('', drop_slot_res.splitlines()[2].strip(),
                         '执行失败:' + text)
        self.assertEqual(len(result1), 0, '执行失败:' + text)
        self.assertEqual(len(result2), 0, '执行失败:' + text)
        self.assertTrue(guc_res1, "执行失败" + text)
        self.assertTrue(guc_res2, "执行失败" + text)
        self.assertTrue(guc_res3, "执行失败" + text)
        self.assertTrue(guc_res4, "执行失败" + text)
        self.assertTrue(guc_res5, "执行失败" + text)
